package com.basic.threadpool;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class SynchronousQueueThreadPoolTest {

	
	public static void main(String[] args) throws InterruptedException {
		Executors.newCachedThreadPool();
		final CountDownLatch countDownLatch = new CountDownLatch(10);
//		final AtomicInteger atomicInt = new AtomicInteger(0);
		SynchronousQueue<Runnable> workQueue = new SynchronousQueue<>(true);
		final ThreadPoolExecutor executor = new ThreadPoolExecutor(0,
				Integer.MAX_VALUE, 
				60, 
				TimeUnit.MINUTES, 
				workQueue, 
				new TestThreadFactory(), 
				new PrintRejectedExecutionHandler());
		long beginTime = System.currentTimeMillis();
		//添加任务
		for (int i = 1; i <= 10; i++) {
			executor.execute(new WorkerTask(i,countDownLatch));
		}
		
		//循环添加任务
//		new Thread(new Runnable() {
//			@Override
//			public void run() {
//				while (true) {
//					for (int i = 1; i <= 4; i++) {
//						executor.execute(new WorkerTask(i,countDownLatch));
//					}
//					try {
//						Thread.sleep(5000);
//					} catch (InterruptedException e) {
//						e.printStackTrace();
//					}
//					
//				}
//			}
//		}).start();
		//查看执行线程数
//		while (true) {
//			try {
//				Thread.sleep(500);
//				System.out.println("执行时线程数： "+ executor.getPoolSize());
//			} catch (InterruptedException e) {
//				e.printStackTrace();
//			}
//		}
		countDownLatch.await();
		long endTime = System.currentTimeMillis();
		System.out.println("任务执行总时间   = "+ (endTime-beginTime));
		executor.shutdown();
//		executor.shutdownNow();
//		try {
//			executor.awaitTermination(1, TimeUnit.SECONDS);
//			
//		} catch (InterruptedException e) {
//			e.printStackTrace();
//		}
	}
	
	public static class WorkerTask implements Runnable{
		
		private CountDownLatch countDownLatch;
		
		private final int taskNum;
		
		public WorkerTask(int taskNum,CountDownLatch countDownLatch){
			this.taskNum = taskNum;
			this.countDownLatch = countDownLatch;
		}
		
		public int getTaskNum() {
			return taskNum;
		}
		
		public void run() {
			try {
				Thread.sleep(1000);
				System.out.println("线程： "+ Thread.currentThread().getName()+" 任务编号："+taskNum+" 任务执行完成......");
				countDownLatch.countDown();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	public static class TestThreadFactory implements ThreadFactory{
		
		private static final AtomicInteger ThreadNum = new AtomicInteger();
		
		public Thread newThread(Runnable r) {
			Thread thread = new Thread(r);
			thread.setName("TEST-POOL-"+ThreadNum.incrementAndGet());
			return thread;
		}
	}
	
	public static class PrintRejectedExecutionHandler implements RejectedExecutionHandler{
		
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			System.out.println("任务被拒绝=========="+ ((WorkerTask)r).getTaskNum()+"===========");
			//重新加入到队列中进行执行。
			try {
				executor.getQueue().put(r);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}
